package chapter_03

import org.apache.spark.{SparkConf, SparkContext}


class chapter_s03 {
  val conf = new SparkConf().setMaster("local").setAppName("My App")
  val sc = new SparkContext(conf)
  sc.setLogLevel("WARN")

  def readLogfile(): Unit = {
    val inputRDD = sc.textFile("pom.xml")
    // 转化操作
    val errorsRDD = inputRDD.filter(line => line.contains("dependency"))
    val warningsRDD = inputRDD.filter(line => line.contains("groupId"))
    val badLinesRDD = errorsRDD.union(warningsRDD)

    // 行动操作
    println("Input had " + badLinesRDD.count() + " concerning lines")
    println("Here are 10 examples:")
    badLinesRDD.take(10).foreach(println)
  }

  def compareWithMapAndFlatMap(): Unit = {
    // 比较map和flatMap
    val rdd = sc.parallelize(
      List("coffee panda", "happy panda", "happiest panda party"))
    val mapResult = rdd.map(x => x.split(" ")).collect()
    println("map的输出结果：")
    for (x <- mapResult) {
      for (r <- x)
        print(r + " ")
      println()
    }

    // 拍扁的效果
    val flatMapResult = rdd.flatMap(x => x.split(" ")).collect()
    println("flatMap的输出结果：")
    flatMapResult.foreach(println)
  }

  /*
  演示对一个数据为{1, 2, 3, 3}的RDD进行基本的RDD转化操作
   */
  def showDirect(): Unit = {
    val rdd = sc.parallelize(List(1, 2, 3, 3))
    // map
    val mapRes = rdd.map(x => x + 1).collect()
    print("map result: ")
    println(mapRes.mkString(", "))

    // flatMap
    val flatMapRes = rdd.flatMap(x => x.to(3))
    print("flatMap result: ")
    flatMapRes.foreach(x => print(x + " "))
    println("")

    // filter
    val filterRes = rdd.filter(x => x != 1).collect()
    print("filter: ")
    println(filterRes.mkString(", "))

    // distinct去重
    val distinctRes = rdd.distinct().collect()
    print("distinct: ")
    println(distinctRes.mkString(", "))

    // sample采样，以及是否替换
    val sampleRes = rdd.sample(false, 0.5).collect()
    print("sample: ")
    println(sampleRes.mkString(", "))
  }

  /*
  对数据分别为{1, 2, 3}和{3, 4, 5}的RDD进行针对两个RDD的转化操作
   */
  def showDirectTwo(): Unit = {
    val rdd1 = sc.parallelize(List(1, 2, 3))
    val rdd2 = sc.parallelize(List(3, 4, 5))

    print("union: ")
    rdd1.union(rdd2).foreach(x => print(x + " "))
    println("")

    print("intersection(共同元素): ")
    rdd1.intersection(rdd2).foreach(x => print(x + " "))
    println("")

    print("subtract(移除一个RDD中的内容): ")
    rdd1.subtract(rdd2).foreach(x => print(x + " "))
    println("")

    print("cartesian(笛卡尔积): ")
    val cartesianRes = rdd1.cartesian(rdd2)
    cartesianRes.foreach(x => print(x + " "))
  }

  // 对一个数据为{1, 2, 3, 3}的RDD进行基本的RDD行动操作
  def showAction(): Unit = {
    val rdd = sc.parallelize(List(1, 2, 3, 3))

    // count
    print("count: ")
    println(rdd.count())

    // countByValue
    print("countByValue(各个元素在RDD中出现的次数): ")
    rdd.countByValue().foreach(x => print(x + " "))
    println("")

    // take
    print("take: ")
    rdd.take(2).foreach(x => print(x + " "))
    println("")

    // top
    print("top: ")
    rdd.top(2).foreach(x => print(x + " "))
    println("")

    // takeOrdered
    print("takeOrdered: ")
    rdd.takeOrdered(2).foreach(x => print(x + " ")) // 从低到高返回指定个数
    println()

    // takeSample
    print("takeSample: ")
    rdd.takeSample(false, 1).foreach(x => print(x + " "))
    println()

    // reduce
    print("reduce: ")
    val reduceRes = rdd.reduce((x, y) => x + y)
    println(reduceRes)

    // fold
    print("fold: ")
    val foldRes = rdd.fold(0)((x, y) => x + y)
    println(foldRes)

    // aggregate
    // aggregate和reduce()相似，但通常返回不同类型的函数
    /*
    def aggregate[U: ClassTag](zeroValue: U)
    (seqOp: (U, T) => U, combOp: (U, U) => U): U = withScope {...}
    将每个分区里面的元素进行聚合，然后用combOp函数将每个分区的结果和初始值
    （zeroValue）进行combOp操作。这个函数最终返回的类型不需要和RDD中元素类型一致。
    seqOp操作会聚合各分区中的元素，然后combOp操作把所有分区的聚合结果再次聚合，两个
    操作的初始值都是zeroValue。seqOp的操作是遍历分区中的所有元素T，第一个T跟zeroValue
    做操作，结果再作为与第二个T做操作的zeroValue，直到遍历完整个分区。combOp操作是把各
    分区聚合的结果再聚合。aggregate函数返回一个跟RDD不同类型的值。因此，需要一个操作
    seqOp来把分区中的元素T合并成一个U，另一个操作combOp把所有U聚合。
     */
    print("aggregate: ")
    val rdd2 = sc.parallelize(1 to 9)
    val aggregateRes = rdd2.aggregate((0, 0))(
      // acc指的是开始自己定义的(0, 0)
      (acc, number) => (acc._1 + number, acc._2 + 1),         // 具体操作
      (par1, par2) => (par1._1 + par2._1, par1._2 + par2._2)  // 分布式合并
    )
    println(aggregateRes._1 / aggregateRes._2)
    /*
    过程大概这样：
    首先，初始值是(0, 0)，这个值在后面2步会用到。
    然后，(acc, number) => (acc._1 + number, acc._2 + 1)，number即是函数定义中的T，这里
    即是List中的元素。所以acc._1 + number, acc._2 + 1的过程如下：
    1.  0+1,  0+1
    2.  1+2,  1+1
    3.  3+3,  2+1
    4.  6+4,  3+1
    5.  10+5,  4+1
    6.  15+6,  5+1
    7.  21+7,  6+1
    8.  28+8,  7+1
    9.  36+9,  8+1
    实际spark执行的是分布式计算，可能会把List分成多个分区，假如3个，p1(1, 2, 3, 4)，
    p2(5, 6, 7, 8)， p3(9)，经过计算各分区的结果(10, 4)， (26, 4)， (9, 1)，这样，
    执行(par1, par2) => (par1._1 + par2._1, par1._2 + par2._2)就是(10 + 26 + 9,
    4 + 4 + 1)即(45, 9)，在再求平均值就简单了。
     */
  }
}


object chapter_s03 {

  def main(args: Array[String]): Unit = {
    val ob = new chapter_s03
//    ob.readLogfile()
//    ob.compareWithMapAndFlatMap()
//    ob.showDirect()
//    ob.showDirectTwo()
    ob.showAction()
  }
}
